package cn.ac.iie.di.ban.data.exchange.runner.slave.task.runTask;

import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.StatusEnum;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.task.order.TaskOrderRequest;
import cn.ac.iie.di.ban.data.exchange.runner.slave.DownloadSlave;
import cn.ac.iie.di.commons.util.hdfs.HDFSUtil;
import cn.iie.jsqlparser.helper.JSQLParserHelper;
import cn.iie.jsqlparser.helper.data.CondTreeData;
import cn.iie.jsqlparser.helper.data.SQLOperatorEnum;
import iie.mm.client.ClientAPI;

import java.io.File;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.util.List;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by Black-Laptop on 12/6/2017.
 */
public class DownloadTaskThread extends Thread {

    private static final Logger LOG = LoggerFactory.getLogger(DownloadTaskThread.class);
    private Boolean stopThread;
    private CondTreeData tree;
    private String resultPath;

    private String MMUrls;
    private String MMnameBase;
    private String sql;
    private JSQLParserHelper jsqlParserHelper;
    private String value;

    private String srcPath;
    private String hdfsPath;

    private String ftpIp;
    private String ftpUser;
    private String ftpPassword;
    private int ftpPort;

    public DownloadTaskThread(TaskOrderRequest taskOrderRequest, JSQLParserHelper jsqlParserHelper, String MMUrls, String MMnameBase, String value) throws Exception {
        this.sql = jsqlParserHelper.getSQLString();
        this.MMUrls = MMUrls;
        this.MMnameBase = MMnameBase;
        this.tree = jsqlParserHelper.getSQLWhere();
        this.jsqlParserHelper = jsqlParserHelper;
        this.hdfsPath = taskOrderRequest.getResult_path();
        this.value = value;
        this.srcPath = "D:\\a\\c\\";
        stopThread = false;
    }


    @Override
    public void run() {
        if (!stopThread) {
            /**
             * 用于where语句的过滤，非"="则直接抛出异常，只执行"="的语句
             */
            SQLOperatorEnum compareOperator = tree.getValue().getOperator();

            LOG.info("sql:" + sql + "  operator=" + compareOperator);

            switch (compareOperator) {
                case EqualsTo: {
                    try {
                        download();
                    } catch (Exception e) {
                        DownloadSlave.callAfterFinished.finished(StatusEnum.DOWNLOAD_TASK_ERROR, ExceptionUtils.getFullStackTrace(e));
                        LOG.error(ExceptionUtils.getFullStackTrace(e));
                    }
                    break;
                }
                default: {
                    DownloadSlave.callAfterFinished.finished(StatusEnum.DOWNLOAD_TASK_ERROR, "parser error:  compareOperator=" + compareOperator);
                    LOG.error("parser error:  compareOperator=" + compareOperator);
                }
            }
        }
        DownloadSlave.callAfterFinished.finished(StatusEnum.DOWNLOAD_TASK_SUCCESS, sql);

    }

    /**
     * 下载从多媒体库中获得的文件/图片等 文件名为key值
     */
    public void download() throws Exception {
        List<String> keys = jsqlParserHelper.getSQLWhereValues();

        LOG.info("key=" + keys.get(0) + "   resultPath=" + resultPath);

        ClientAPI ca = new ClientAPI();
        ca.init(MMUrls, MMnameBase);
        //从偏移offset读取长度为length的数据,若一次读取整个小文件则把len参数设为-1

        String[] s = value.split("/");
        String fileName = s[s.length - 1];

        File srcdirs = new File(srcPath);
        dirExists(srcdirs);
        File srcFile = new File(srcPath + fileName);
        fileExists(srcFile);

        for (String key : keys) {
            byte[] content = ca.downloadFile(key, 0, -1);

            //写数据
            OutputStream os = new FileOutputStream(srcFile);
            os.write(content, 0, content.length);
        }

        ca.quit();
        try {
            HDFSUtil.putFileFromLocal(srcPath + fileName, hdfsPath);
            srcFile.delete();
        } catch (Exception ex) {
            LOG.error("upload the operator registration package to HDFS failed. Case:" + ex.getMessage(), ex);
        }

    }

    private void dirExists(File file) {
        if (file.exists()) {
            if (file.isDirectory()) {
//                System.out.println("dir exists");
            } else {
                System.out.println("same name file exists");
            }
        } else {
//            System.out.println("dir not exists");
            file.mkdirs();
        }
    }

    private void fileExists(File file) {
        if (!file.exists()) {
            try {
                file.createNewFile();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }

    }

    public void stopThrad() {
        this.stopThread = true;
    }
}
